package com.we.flink.accountuserlabelplatform.tag;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.we.flink.accountuserlabelplatform.table.TableRepayPlan;
import com.we.flink.mobilelabelplatform.tag.TagSpecialContact;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.util.Iterator;

public class TagRepayPlan {
    public String account_user_key;
    public Integer earliest_due_date;
    public BigDecimal overdue_principal;

    public static Logger LOG = LoggerFactory.getLogger(TagSpecialContact.class);
    public static final BigDecimal bigdecimal0 = new BigDecimal(String.valueOf(0));
    public static final String SEPARATOR = "#";

    public TagRepayPlan() {}

    public TagRepayPlan(
            String account_user_key, Integer earliest_due_date, BigDecimal overdue_principal) {
        this.account_user_key = account_user_key;
        this.earliest_due_date = earliest_due_date;
        this.overdue_principal = overdue_principal;
    }

    public static TagRepayPlan init(String user_key) {
        return new TagRepayPlan(user_key, Integer.MAX_VALUE, bigdecimal0);
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableRepayPlan.QUERYSQL);
        DataStream<Tuple2<Boolean, TableRepayPlan>> inputStream =
                tableEnv.toRetractStream(tableRes, TableRepayPlan.class);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .filter(
                                new FilterFunction<Tuple2<Boolean, TableRepayPlan>>() {
                                    @Override
                                    public boolean filter(Tuple2<Boolean, TableRepayPlan> value)
                                            throws Exception {
                                        return value.f0;
                                    }
                                })
                        .keyBy(
                                new KeySelector<Tuple2<Boolean, TableRepayPlan>, String>() {
                                    @Override
                                    public String getKey(Tuple2<Boolean, TableRepayPlan> value)
                                            throws Exception {
                                        return value.f1.getAccount_user_key();
                                    }
                                })
                        .flatMap(new TagRepayPlanFlatMapFunc());

        return out;
    }

    public String getAccount_user_key() {
        return account_user_key;
    }

    public Integer getEarliest_due_date() {
        return earliest_due_date;
    }

    public BigDecimal getOverdue_principal() {
        return overdue_principal;
    }

    private static class OverduePeriodInfo {
        public int due_date;
        public float left_principal;

        public OverduePeriodInfo(int due_date, float left_principal) {
            this.due_date = due_date;
            this.left_principal = left_principal;
        }

        public int getDue_date() {
            return due_date;
        }

        public float getLeft_principal() {
            return left_principal;
        }
    }

    /** 带loankey+period 明细的计算 */
    private static class TagRepayPlanFlatMapFunc
            extends RichFlatMapFunction<Tuple2<Boolean, TableRepayPlan>, String> {
        MapState<String, String> repayPlanMapState;
        ValueState<TagRepayPlan> repayPlanVState;

        @Override
        public void open(Configuration parameters) throws Exception {
            repayPlanMapState =
                    getRuntimeContext()
                            .getMapState(
                                    new MapStateDescriptor<String, String>(
                                            "RepayPlanMapState", String.class, String.class));
            repayPlanVState =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<TagRepayPlan>(
                                            "RepayPlanVState", TagRepayPlan.class));
        }

        @Override
        public void flatMap(Tuple2<Boolean, TableRepayPlan> value, Collector<String> out)
                throws Exception {
            String account_user_key = value.f1.getAccount_user_key();
            String loan_key = value.f1.getLoan_key();
            //            LOG.error("##dbg jk account_user_key: " + account_user_key + " loan_key: "
            // + loan_key + " period:" + value.f1.getPeriods());
            //            System.out.println("##dbg jk account_user_key: " + account_user_key + "
            // loan_key: " + loan_key + " period:" + value.f1.getPeriods());
            int period = value.f1.getPeriods().intValue();
            String status = value.f1.getStatus();
            Timestamp due_date = value.f1.getDue_date();
            int duedate_time = (int) (due_date.getTime() / 1000);
            BigDecimal left_principal = value.f1.getLeft_principal();

            //            System.out.println("##dbg act_u_k: " + account_user_key +
            //                    " loan_key: " + loan_key +
            //                    " period: " + period +
            //                    " status: " + status +
            //                    " due_date: " + due_date +
            //                    " dueDate: "+ duedate_time +
            //                    " leftPrincipal: " + left_principal );
            String state_key = loan_key + SEPARATOR + period;
            String state_val = status + SEPARATOR + duedate_time + SEPARATOR + left_principal;

            //            System.out.println("##dbg state_key: " + state_key + " state_val: " +
            // state_val);

            TagRepayPlan tmpState = repayPlanVState.value();
            TagRepayPlan curState =
                    tmpState == null ? TagRepayPlan.init(account_user_key) : tmpState;
            Integer min_due_date = Integer.MAX_VALUE;
            BigDecimal sum_left_principal = bigdecimal0;

            /** 能进来代表有记录，至少输出0，0 */
            repayPlanMapState.put(state_key, state_val);

            if (!repayPlanMapState.isEmpty()) {
                Iterator<String> period_info = repayPlanMapState.values().iterator();

                while (period_info.hasNext()) {
                    String info = period_info.next();
                    //                    System.out.println("##dbg period_info: "+ info);
                    if (info.contains("OVERDUE")) {
                        String[] splits = info.split(SEPARATOR);
                        if (splits.length == 3) {
                            int tmp_due_date = Integer.valueOf(splits[1]).intValue();
                            BigDecimal tmp_left_pricipal = new BigDecimal(splits[2]);
                            //                            System.out.println("##dbg OVERDUE:
                            // tmp_due_date: " + tmp_due_date + " tmp_left_pricipal: " +
                            // tmp_left_pricipal);

                            if (min_due_date > tmp_due_date) {
                                min_due_date = tmp_due_date;
                            }
                            sum_left_principal = sum_left_principal.add(tmp_left_pricipal);
                        }
                    }
                }
                //                System.out.println("##dbg curstate due_date:" +
                // curState.getEarliest_due_date()  + " curstate left_principal: " +
                // curState.getOverdue_principal());
                //                System.out.println("##dbg min_due_date: " + min_due_date + "
                // sum_left_principal: " + sum_left_principal);
                /** 如果和状态记录不一致，需要更新，输出计算出来的当前最早应还款日和 逾期本金和 如果和初始状态记录一致，表示当前无逾期记录或者都已还清，输出0，0 */
                if (!curState.getEarliest_due_date().equals(min_due_date)
                        || !curState.getOverdue_principal().equals(sum_left_principal)
                        || (min_due_date.equals(Integer.MAX_VALUE)
                                && sum_left_principal.equals(bigdecimal0))) {
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("account_user_key", account_user_key);
                    jsonObject.put("overdue_principal", sum_left_principal.floatValue());
                    /** 如果和初始状态应还款日一致，输出0 */
                    if (min_due_date.equals(Integer.MAX_VALUE)) {
                        jsonObject.put("earliest_due_date", Integer.valueOf(0));
                    } else {
                        jsonObject.put("earliest_due_date", min_due_date);
                    }

                    //                    System.out.println("##dbg collect: " +
                    // jsonObject.toString());
                    out.collect(jsonObject.toString());

                    curState.earliest_due_date = min_due_date;
                    curState.overdue_principal = sum_left_principal;
                    repayPlanVState.update(curState);
                } else {
                    //                    System.out.println("##dbg not update!!");
                }
            }
        }
    }
}
